package io.dy;

import io.dy.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author : dmj
 * @date: 2024/8/27 11:25 上午
 * @description: 消息重试
 *
 * 1、什么情况会触发重试？
 *      监听器返回null || RECONSUME_LATER 会触发重试
 * 2、重试机制是什么？
 *      默认消费者不设置重试次数 那么（并发模式-MessageListenerConcurrently）会重试16次 时间间隔 与 延时消息等级相同，顺序模式下重试int的最大值
 * 3、重试次数用尽还是失败消息会去哪里？
 *      会进入死信主题中 名字规律 %DLQ% + 消费者组名称 ===> %DLQ%retry-consumer-group
 */
public class RocketMqRetry {


    @Test
    public void producer() throws Exception {

        //1、创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("retry-producer-group");

        //2、连接namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、启动
        producer.start();

        //4、发消息
        Message message = new Message("retryTopic", "retry消息77777".getBytes());
        SendResult sendResult = producer.send(message);
        System.out.println(sendResult.getSendStatus());

        //5、关闭
        producer.shutdown();
    }


    /**
     * 未设置重试次数
     * 1、什么情况会触发重试？
     *      监听器返回null || RECONSUME_LATER 会触发重试
     * 2、重试机制是什么？
     *      默认消费者不设置重试次数 那么会重试16次 时间间隔 与 延时消息等级相同
     * @throws Exception
     */
    @Test
    public void consumer() throws Exception {

        //1、创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");

        //2、连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、订阅主题    * 表示订阅主题中的所有消息
        consumer.subscribe("retryTopic", "*");

        //4、设置监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = list.get(0);
                System.out.println("时间: " + new Date());
                System.out.println("重试次数: " + messageExt.getReconsumeTimes());
                System.out.println(new String(messageExt.getBody()));
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        //5、启动
        consumer.start();

        //6、阻塞
        new CountDownLatch(1).await();
    }

    /**
     * 设置重试次数
     * 1、什么情况会触发重试？
     *      监听器返回null || RECONSUME_LATER 会触发重试
     * 2、重试机制是什么？
     *      默认消费者不设置重试次数 那么会重试16次 时间间隔 与 延时消息等级相同
     * @throws Exception
     */
    @Test
    public void retryConsumer() throws Exception {

        //1、创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-consumer-group");

        //2、连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、订阅主题    * 表示订阅主题中的所有消息
        consumer.subscribe("retryTopic", "*");

        //设置重试次数
        consumer.setMaxReconsumeTimes(1);

        //4、设置监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = list.get(0);
                System.out.println("时间: " + new Date());
                System.out.println("重试次数: " + messageExt.getReconsumeTimes());
                System.out.println(new String(messageExt.getBody()));
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        //5、启动
        consumer.start();

        //6、阻塞
        new CountDownLatch(1).await();
    }


    /**
     * 监听死信主题
     * @throws Exception
     */
    @Test
    public void dlqConsumer() throws Exception {

        //1、创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("retry-dlq-consumer-group");

        //2、连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDRESS);

        //3、订阅主题    * 表示订阅主题中的所有消息
        consumer.subscribe("%DLQ%retry-consumer-group", "*");

        //设置重试次数
        consumer.setMaxReconsumeTimes(1);

        //4、设置监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                MessageExt messageExt = list.get(0);
                System.out.println("时间: " + new Date());
                System.out.println(new String(messageExt.getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //5、启动
        consumer.start();

        //6、阻塞
        new CountDownLatch(1).await();
    }
}
